package com.sali.service.statistic;


import com.sali.commons.config.TaskConfig;
import com.sali.commons.config.ThreadPoolManager;
import com.sali.entity.StatisticalIndicator;
import com.sali.entity.TransferObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

// 统计指标收集器
public class StatisticalIndicatorCollector {

    private static final Logger log = LoggerFactory.getLogger(StatisticalIndicatorCollector.class);

    // 统计各个类型的指标
    private static final Map<String, StatisticalIndicator> protocolCount = new ConcurrentHashMap<>();

    // 前一天的统计指标
    private static final Map<String, StatisticalIndicator> yesterdayProtocolCount = new HashMap<>();

    private static final long printInterval;

    static {
        printInterval = TaskConfig.STATISTIC_LOG_PRINT_INTERVAL;

        // 定时打印统计日志
        ThreadPoolManager.getScheduledExecutor().scheduleAtFixedRate(StatisticalIndicatorCollector::print, 0, printInterval, TimeUnit.SECONDS);
    }

    public static void print() {
        // 计算总数
        long inputTotal = 0L;

        // 计算输出总数
        long outputTotal = 0L;

        // 计算异常总数
        long errorTotal = 0L;

        // 计算过滤总数
        long filterTotal = 0L;

        // 计算最近5秒的总数
        long lastNSecInputTotal = 0L;

        // 计算最近5s输出处理总数
        long lastNSecOutputTotal = 0L;

        // 计算最近5s 异常总数
        long lastNSecErrorTotal = 0L;

        // 计算最近5s过滤总数
        long lastNSecFilterTotal = 0L;

        // 计算最近5s的平均输入、输出速率,基于lastNSecInputTotal、lastNSecOutputTotal计算
        double lastNSecInputSpeed;
        double lastNSecOutputSpeed;

        for (Map.Entry<String, StatisticalIndicator> entity : protocolCount.entrySet()) {
            StatisticalIndicator indicator = entity.getValue();
            inputTotal += indicator.getInputCount().get();
            outputTotal += indicator.getOutputCount().get();
            errorTotal += indicator.getErrorCount().get();
            filterTotal += indicator.getFilterCount().get();
            lastNSecInputTotal += (indicator.getInputCount().get() - indicator.getLastInputTotal().get());
            lastNSecErrorTotal += (indicator.getErrorCount().get() - indicator.getLastErrorTotal().get());
            lastNSecFilterTotal += (indicator.getFilterCount().get() - indicator.getLastFilterTotal().get());
            lastNSecOutputTotal += (indicator.getOutputCount().get() - indicator.getLastOutputTotal().get());

            //10 输出各个协议的上述9项指标
            boolean printDetail = TaskConfig.STATISTIC_LOG_DETAIL;
            if (printDetail) {
                double singleProtocolInputSpeed = (double) (indicator.getInputCount().get() - indicator.getLastInputTotal().get()) / printInterval;
                double singleProtocolOutputSpeed = (double) (indicator.getOutputCount().get() - indicator.getLastOutputTotal().get()) / printInterval;
                String logMsg1 = String.format("[数据统计] 协议类型:%s,总接入条数：%d，总输出条数：%d，总异常条数：%d,总过滤条数：%d，最近%d秒接入总数：%d，最近%d秒输出总数：%d，最近%d秒异常总数：%d，最近%d秒过滤总数:%d，最近%d秒输入速率：%.2f条/秒，最近%d秒输出速率：%.2f条/秒",
                        entity.getKey(), indicator.getInputCount().get(), indicator.getOutputCount().get(), indicator.getErrorCount().get(), indicator.getFilterCount().get(),
                        printInterval, indicator.getInputCount().get() - indicator.getLastInputTotal().get(),
                        printInterval, indicator.getOutputCount().get() - indicator.getLastOutputTotal().get(),
                        printInterval, indicator.getErrorCount().get() - indicator.getLastErrorTotal().get(),
                        printInterval, indicator.getFilterCount().get() - indicator.getLastFilterTotal().get(),
                        printInterval, singleProtocolInputSpeed,
                        printInterval, singleProtocolOutputSpeed);
                log.info(logMsg1);

            }

            //重置指标,开始下一轮统计
            indicator.updateLastRecord();

        }

        lastNSecInputSpeed = (double) lastNSecInputTotal / printInterval;
        lastNSecOutputSpeed = (double) lastNSecOutputTotal / printInterval;

        String logMsg2 = String.format("[数据统计] 总接入条数：%d，总输出条数：%d，总异常条数：%d，总过滤条数：%d，最近%d秒接入总数：%d，最近%d秒输出总数：%d，最近%d秒异常总数：%d，最近%d秒过滤总数：%d，最近%d秒输入速率：%.2f条/秒，最近%d秒输出速率：%.2f条/秒",
                inputTotal, outputTotal, errorTotal, filterTotal,
                printInterval, lastNSecInputTotal,
                printInterval, lastNSecOutputTotal,
                printInterval, lastNSecErrorTotal,
                printInterval, lastNSecFilterTotal,
                printInterval, lastNSecInputSpeed,
                printInterval, lastNSecOutputSpeed);
        log.info(logMsg2);

    }

    /**
     * 收集输入条数指标
     */
    public static void inputInc(TransferObject val) {
        CompletableFuture.runAsync(() -> {
            try {
                String protocolCode = val.getProtocolCode();
                protocolCount.computeIfAbsent(protocolCode, key -> new StatisticalIndicator()).inputInc();

            } catch (Exception e) {
                String errorMessage = String.format("[数据统计] 指标统计异常，data:%s, error msg: %s", val.getData(), e.getMessage());
                log.error(errorMessage, e);
            }
        });
    }

    /**
     * 收集输出条数指标
     */
    public static void outputInc(TransferObject val) {
        CompletableFuture.runAsync(() -> {
            try {
                String protocolCode = val.getProtocolCode();
                protocolCount.computeIfAbsent(protocolCode, key -> new StatisticalIndicator()).outputInc();

            } catch (Exception e) {
                String errorMessage = String.format("[数据统计] 指标统计异常，data:%s, error msg: %s", val.getData(), e.getMessage());
                log.error(errorMessage, e);
            }
        });
    }


    /**
     * 收集过滤条数指标
     */
    public static void filterInc(TransferObject val) {
        CompletableFuture.runAsync(() -> {
            try {
                String protocolCode = val.getProtocolCode();
                protocolCount.computeIfAbsent(protocolCode, key -> new StatisticalIndicator()).filterCountInc();

            } catch (Exception e) {
                String errorMessage = String.format("[数据统计] 指标统计异常，data:%s, error msg: %s", val.getData(), e.getMessage());
                log.error(errorMessage, e);
            }
        });
    }


    /**
     * 收集异常条数指标
     */
    public static void errorInc(TransferObject val) {
        CompletableFuture.runAsync(() -> {
            try {
                String protocolCode = val.getProtocolCode();
                protocolCount.computeIfAbsent(protocolCode, key -> new StatisticalIndicator()).errorCountInc();

            } catch (Exception e) {
                String errorMessage = String.format("[数据统计] 指标统计异常，data:%s, error msg: %s", val.getData(), e.getMessage());
                log.error(errorMessage, e);
            }
        });
    }

}
